package cn.bby.query.server.infrastructure.impl.storage;//package cn.bby.query.server.service.storage;
//
//import cn.bby.query.base.constant.AdvQueryErrorCode;
//import cn.bby.query.base.constant.ColumnType;
//import cn.bby.query.base.constant.QueryCondition;
//import cn.bby.query.base.exception.AdvQueryException;
//import cn.bby.query.base.exception.AdvQueryLoadException;
//import cn.bby.query.base.model.req.Filter;
//import cn.bby.query.base.model.req.QueryRequest;
//import cn.bby.query.base.model.req.QueryResponse;
//import cn.bby.query.server.domain.metadata.model.dto.ColumnInfoDTO;
//import cn.bby.query.server.model.dto.DataUpdateDetail;
//import cn.bby.query.server.domain.metadata.model.dto.InterfaceInfo;
//import cn.bby.query.server.infrastructure.db.entity.InterfaceInfoPO;
//import cn.bby.query.server.util.ErrorLogUtil;
//import cn.hutool.core.collection.CollUtil;
//import co.elastic.clients.elasticsearch.ElasticsearchClient;
//import co.elastic.clients.elasticsearch._types.FieldValue;
//import co.elastic.clients.elasticsearch._types.SortOptions;
//import co.elastic.clients.elasticsearch._types.SortOrder;
//import co.elastic.clients.elasticsearch._types.mapping.Property;
//import co.elastic.clients.elasticsearch._types.query_dsl.Query;
//import co.elastic.clients.elasticsearch.core.BulkRequest;
//import co.elastic.clients.elasticsearch.core.BulkResponse;
//import co.elastic.clients.elasticsearch.core.UpdateByQueryResponse;
//import co.elastic.clients.elasticsearch.core.bulk.BulkOperation;
//import co.elastic.clients.elasticsearch.indices.CreateIndexRequest;
//import co.elastic.clients.elasticsearch.indices.CreateIndexResponse;
//import co.elastic.clients.elasticsearch.indices.ElasticsearchIndicesClient;
//import co.elastic.clients.transport.endpoints.BooleanResponse;
//import com.alibaba.fastjson2.JSONObject;
//import lombok.Data;
//import lombok.RequiredArgsConstructor;
//import lombok.extern.slf4j.Slf4j;
//import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
//import org.springframework.stereotype.Service;
//
//import java.io.IOException;
//import java.util.*;
//import java.util.stream.Collectors;
//
/// **
// * ES 数据存储管理器
// *
// * @author ben
// * @since 2025/4/19 23:48
// */
//@Service
//@Slf4j
//@RequiredArgsConstructor
//@ConditionalOnProperty(name = "query.storage.manager", havingValue = "es")
//public class EsDataStorageManagerImpl implements DataStorageManager {
//
//    private final ElasticsearchClient elasticsearchClient;
//    private final ErrorLogUtil errorLogUtil;
//
//    Map<String, ConditionHandler> conditionMap = Map.ofEntries(
//            Map.entry(QueryCondition.EQUALS, (Query.Builder queryBuilder, FieldValueFilter filter) -> {
//                if (filter.getValue().size() == 1) {
//                    queryBuilder.term(t -> t
//                            .field(filter.getField())
//                            .value(filter.getFieldValues().get(0))
//                    );
//                } else {
//                    queryBuilder.terms(t -> t
//                            .field(filter.getField())
//                            .terms(term -> term.value(filter.getFieldValues()))
//                    );
//                }
//            })
//    );
//
//    private static Property.Builder handleColumn(ColumnInfoDTO column, Property.Builder pb) {
//        switch (ColumnType.getByCode(column.getType())) {
//            case NUMBER:
//                pb.long_(p -> p);
//                break;
//            case STRING:
//                pb.text(p -> p);
//                break;
//            case DATE:
//                pb.date(p -> p.format("yyyy-MM-dd  HH:mm:ss"));
//                break;
//            case BOOLEAN:
//                pb.boolean_(p -> p);
//                break;
//            case LIST:
//                pb.text(o -> o);
//                break;
//            default:
//                throw new AdvQueryLoadException("未知类型:" + column.getType());
//        }
//        return pb;
//    }
//
//    private static FieldValue castFieldValue(String type, String value) {
//        switch (ColumnType.getByCode(type)) {
//            case NUMBER:
//                return FieldValue.of(Long.parseLong(value));
//            case BOOLEAN:
//                return FieldValue.of(Boolean.parseBoolean(value));
//            default:
//                return FieldValue.of(value);
//        }
//    }
//
//    private static String getIndexName(InterfaceInfoPO interfaceInfo) {
//        return interfaceInfo.getInterfaceName().toLowerCase();
//    }
//
//    @Override
//    public void finishLoad(InterfaceInfoPO interfaceInfo) {
//        log.warn("finishAsync:{}", interfaceInfo);
//    }
//
//    @Override
//    public void initDataStruct(InterfaceInfo interfaceInfo) {
//        log.warn("initDataStruct:{}", interfaceInfo);
//        try {
//            BooleanResponse exists = elasticsearchClient.indices().exists(d -> d.index(getIndexName(interfaceInfo)));
//            if (exists.value()) {
//                throw new AdvQueryLoadException("索引已存在，无法重新初始化!");
//            }
//            CreateIndexRequest.Builder builder = new CreateIndexRequest.Builder()
//                    .index(getIndexName(interfaceInfo));
//            String primaryField = interfaceInfo.getPrimaryField();
//            builder.mappings(m -> {
//                        interfaceInfo.getColumns().forEach(column -> m
//                                .properties(column.getLinkCol(), p -> primaryField.equals(column.getFieldName()) ? p.keyword(k -> k) : handleColumn(column, p)));
//                        return m;
//                    }
//            );
//            CreateIndexResponse response = elasticsearchClient.indices().create(builder.build());
//            if (response.acknowledged()) {
//                log.info("创建成功:{}", response.index());
//            } else {
//                throw new AdvQueryLoadException("创建索引失败:" + response);
//            }
//        } catch (Exception e) {
//            throw new AdvQueryLoadException("创建索引失败:", e);
//        }
//    }
//
//    @Override
//    public void saveData(InterfaceInfo interfaceInfo, List<JSONObject> dataSet, String offset) {
//        log.warn("saveData:size={},offset={},interfaceInfo={}", dataSet.size(), offset, interfaceInfo);
//        try {
//            // 批量保存
//            String index = getIndexName(interfaceInfo);
//            BulkRequest.Builder bulkBuilder = new BulkRequest.Builder();
//            for (JSONObject doc : dataSet) {
//                bulkBuilder.operations(op -> op
//                        .create(c -> c
//                                .id(doc.getString(interfaceInfo.getPrimaryField()))
//                                .index(index)
//                                .document(doc)
//                        )
//                );
//            }
//            BulkResponse response = elasticsearchClient.bulk(bulkBuilder.build());
//            if (response.errors()) {
//                throw new AdvQueryLoadException("批量保存失败:" + response);
//            }
//        } catch (Exception e) {
//            throw new AdvQueryLoadException("批量保存失败", e);
//        }
//    }
//
//    @Override
//    public void deleteDataStruct(List<InterfaceInfoPO> interfaceInfos) {
//        // 删除
//        try {
//            ElasticsearchIndicesClient indicesClient = elasticsearchClient.indices();
//            for (InterfaceInfoPO interfaceInfo : interfaceInfos) {
//                if (!indicesClient.exists(d -> d.index(getIndexName(interfaceInfo))).value()) {
//                    return;
//                }
//                indicesClient.delete(d -> d.index(getIndexName(interfaceInfo)));
//            }
//        } catch (IOException e) {
//            throw new AdvQueryLoadException("删除索引失败", e);
//        }
//    }
//
//    @Override
//    public QueryResponse<JSONObject> query(InterfaceInfo interfaceDTO, QueryRequest request) {
//        // String indexName = getIndexName(interfaceDTO);
//        // SearchRequest.Builder searchBuilder = new SearchRequest.Builder()
//        //         .index(indexName)
//        //         .query(q -> buildQuery(interfaceDTO.getColumns(), request, q)) // 构建过滤条件
//        //         .from((request.getPage() - 1) * request.getSize()) // 分页计算
//        //         .size(request.getSize())
//        //         .trackTotalHits(t -> t.enabled(!request.isIgnoreTotal())); // 总数统计控制
//        //
//        // // 添加排序条件
//        // if (CollUtil.isNotEmpty(request.getSorts())) {
//        //     searchBuilder.sort(buildSorts(request.getSorts()));
//        // }
//        //
//        // try {
//        //     SearchResponse<JSONObject> response = elasticsearchClient.search(searchBuilder.build(), JSONObject.class);
//        //     if (response.timedOut()) {
//        //         throw new AdvQueryException(AdvQueryErrorCode.QUERY_TIMEOUT, request);
//        //     }
//        //     HitsMetadata<JSONObject> hits = response.hits();
//        //     if (hits == null) {
//        //         return new QueryResponse<>(request);
//        //     }
//        //     return new QueryResponse<JSONObject>(request)
//        //             .setTotal(hits.total() == null ? 0 : hits.total().subTables())
//        //             .setDataSet(hits.hits().stream().map(Hit::source).collect(Collectors.toList()));
//        // } catch (IOException e) {
//        //     throw new AdvQueryException(AdvQueryErrorCode.QUERY_FAILED, request);
//        // }
//        return new QueryResponse<>(request);
//    }
//
//    @Override
//    public Set<String> queryIdByPrimaryValList(InterfaceInfo interfaceInfo, Map<ColumnInfoDTO, Set<String>> map) {
//        // if (tableId.equals(interfaceInfo.getSourceTableId())) {
//        //     Object val = map.get(interfaceInfo.getPrimaryColumn());
//        //     if (val != null) {
//        //         return Set.of(val.toString());
//        //     }
//        // }
//        // String indexName = getIndexName(interfaceInfo);
//        // // 只查询主键字段
//        // SearchRequest.Builder searchBuilder = new SearchRequest.Builder()
//        //         .index(indexName)
//        //         .source(s -> s.filter(f -> f.includes(interfaceInfo.getPrimaryField())));
//        // List<ColumnInfoDTO> cols = interfaceInfo.getColumns().stream()
//        //         .filter(c -> tableId.equals(c.getTableId()) && map.containsKey(c.getColumnName()))
//        //         .collect(Collectors.toList());
//        // if (CollUtil.isEmpty(cols)) {
//        //     return Collections.emptySet();
//        // }
//        // searchBuilder.query(q -> q.term(t -> {
//        //     cols.forEach(c -> {
//        //                 Object val = map.get(c.getColumnName());
//        //                 String valStr = val == null ? null : val.toString();
//        //                 t.field(c.getFieldName()).subTables(valStr);
//        //             }
//        //     );
//        //     return t;
//        // }));
//        // try {
//        //     SearchResponse<JSONObject> response = elasticsearchClient.search(searchBuilder.build(), JSONObject.class);
//        //     if (response.timedOut()) {
//        //         return Collections.emptySet();
//        //     }
//        //     return response.hits().hits().stream()
//        //             .filter(h -> h.source() != null)
//        //             .map(h -> h.source().getString(interfaceInfo.getPrimaryField()))
//        //             .collect(Collectors.toSet());
//        // } catch (IOException e) {
//        //     return Collections.emptySet();
//        // }
//        return Collections.emptySet();
//    }
//
//    @Override
//    public void updateData(InterfaceInfo interfaceInfo, DataUpdateDetail event) {
//        String index = getIndexName(interfaceInfo);
//        Map<String, Object> data = event.getData();
//        Long tableId = event.getTableId();
//
//        final ColumnInfoDTO keyField = interfaceInfo.getColumns()
//                .stream()
//                .filter(ColumnInfoDTO::getIsPrimary)
//                .findFirst()
//                .orElse(null);
//        if (keyField == null) {
//            log.debug("{}:数据未找到主键:{}", interfaceInfo.getInterfaceName(), event.getData());
//            return;
//        }
//        StringBuilder source = new StringBuilder();
//        for (ColumnInfoDTO column : interfaceInfo.getColumns()) {
//            if (!Objects.equals(column.getTableId(), tableId) || !data.containsKey(column.getLinkCol())) {
//                // 没有的字段，非当前表字段
//                continue;
//            }
//            Object val = data.get(column.getLinkCol());
//
//            source.append("ctx._source.").append(column.getFieldName())
//                    .append(" = ");
//            if (val == null) {
//                source.append("null;");
//                continue;
//            }
//            if (ColumnType.STRING.equals(column.getType())) {
//                source.append("'").append(val).append("';");
//            } else {
//                source.append(val).append(";");
//            }
//        }
//        if (data.get(keyField.getLinkCol()) == null || source.length() == 0) {
//            log.debug("无数关联的据更新:" + event.getData());
//            return;
//        }
//        log.debug("source:{}", source);
//        try {
//            UpdateByQueryResponse response = elasticsearchClient.updateByQuery(builder -> builder
//                    .index(index)
//                    .query(q ->
//                            q.term(t -> t
//                                    .field(keyField.getFieldName())
//                                    .value(Long.parseLong((String) data.get(keyField.getLinkCol())))
//                            )
//                    )
//                    .script(s -> s.source(source.toString()))
//            );
//            if (Boolean.TRUE.equals(response.timedOut())) {
//                errorLogUtil.write(interfaceInfo.getInterfaceId(), "更新超时");
//            }
//            if (response.updated() == null || response.updated() == 0) {
//                log.debug("未匹配到数据:{}", event.getData());
//            } else {
//                log.debug("更新成功:{}", response.updated());
//            }
//        } catch (Exception e) {
//            errorLogUtil.write(interfaceInfo.getInterfaceId(), e);
//        }
//    }
//
//    @Override
//    public void updateDataBatchById(InterfaceInfo interfaceInfo, Set<String> keyValues, List<JSONObject> dataSet) {
//        List<BulkOperation> operations = dataSet.stream()
//                .map(jsonObject -> {
//                    String id = jsonObject.getString(interfaceInfo.getPrimaryField());
//                    return BulkOperation.of(op -> op.update(u -> u
//                            .id(id)
//                            .action(a -> a.doc(jsonObject))
//                    ));
//                })
//                .collect(Collectors.toList());
//        BulkRequest request = new BulkRequest.Builder()
//                .index(getIndexName(interfaceInfo))
//                .operations(operations)
//                .build();
//        try {
//            BulkResponse response = elasticsearchClient.bulk(request);
//            if (response.errors()) {
//                log.error("批量更新失败:{}", response);
//            }
//            log.debug("批量更新成功:{}", response);
//        } catch (IOException e) {
//            throw new RuntimeException(e);
//        }
//    }
//
//    // 构建查询条件
//    private co.elastic.clients.util.ObjectBuilder<Query> buildQuery(List<ColumnInfoDTO> columns
//            , QueryRequest request, Query.Builder queryBuilder) {
//        HashMap<String, String> fieldTypeMap = columns.stream().collect(Collectors.toMap(ColumnInfoDTO::getFieldName,
//                ColumnInfoDTO::getType, (a, b) -> a, HashMap::new));
//        List<Filter> filters = request.getFilters();
//        if (CollUtil.isEmpty(filters)) {
//            return queryBuilder.matchAll(m -> m);
//        }
//
//        for (Filter filter : filters) {
//            String type = fieldTypeMap.get(filter.getField());
//            if (type == null) {
//                throw new AdvQueryException(AdvQueryErrorCode.FIELD_NOT_EXIST, request);
//            }
//            String condition = filter.getCondition();
//            ConditionHandler handler = conditionMap.get(condition);
//            if (handler == null) {
//                throw new AdvQueryException(AdvQueryErrorCode.CONDITION_NOT_SUPPORT, request);
//            }
//            handler.handle(queryBuilder, new FieldValueFilter(filter, type));
//        }
//        return queryBuilder;
//    }
//
//    // 构建排序条件
//    private List<SortOptions> buildSorts(List<QueryRequest.Sort> sorts) {
//        return sorts.stream().map(sort ->
//                new SortOptions.Builder()
//                        .field(f -> f
//                                .field(sort.getField())
//                                .order(SortOrder.Asc.jsonValue().equalsIgnoreCase(sort.getOrder()) ?
//                                        SortOrder.Asc : SortOrder.Desc)
//                        )
//                        .build()
//        ).collect(Collectors.toList());
//    }
//
//    private interface ConditionHandler {
//        void handle(Query.Builder queryBuilder, FieldValueFilter filter);
//    }
//
//    @Data
//    private static class FieldValueFilter {
//        public Filter filter;
//        public String type;
//        public List<FieldValue> fieldValues;
//
//        public FieldValueFilter(Filter filter, String type) {
//            this.filter = filter;
//            this.type = type;
//            this.fieldValues = filter.getValues().stream()
//                    .map(value -> castFieldValue(type, value))
//                    .collect(Collectors.toList());
//        }
//
//        public List<String> getValue() {
//            return filter.getValues();
//        }
//
//        public String getField() {
//            return filter.getField();
//        }
//    }
//}
